# -*- coding: utf-8 -*-
"""
消息中继抽象 & 实现
"""
from __future__ import annotations

import hashlib
import json
from abc import ABC, abstractmethod
from typing import Dict, Any

from common.base_enums import EventType
from common.logger import create_logger

logger = create_logger("domain.relay", "INFO")


class MessageRelay(ABC):
	@abstractmethod
	def relay(self, event_type: str, payload: Dict[str, Any]) -> None:
		raise NotImplementedError


class RedisRelay(MessageRelay):
	"""Demo 阶段：Redis 去重 + WebSocket"""

	def __init__(self, redis_client):
		self.redis = redis_client

	def relay(self, event_type: str, payload: Dict[str, Any]) -> None:
		# 获取告警信息用于日志
		alarm_type = payload.get('alarm_type', 'UNKNOWN')
		meter_code = payload.get('meter_code', 'UNKNOWN')
		alarm_id = payload.get('alarm_id', 'UNKNOWN')

		# 计算指纹：确保同一告警的不同事件类型能被正确区分
		fingerprint_data = {
			"event_type": event_type,  # 关键：事件类型必须参与去重计算
			"alarm_id": alarm_id,  # 核心：告警ID
			"meter_code": meter_code,  # 核心：设备编码
			"alarm_type": alarm_type,  # 核心：告警类型
		}
		fingerprint = hashlib.md5(
			json.dumps(fingerprint_data, sort_keys=True).encode()
		).hexdigest()
		key = f"outbox:processed:{fingerprint}"
		ttl_seconds = 45  # 缩短TTL到45秒

		# 增强日志：记录所有事件的Redis去重检查
		logger.info(f"[RedisRelay] 检查事件: type={event_type}, alarm_id={alarm_id}, meter_code={meter_code}, alarm_type={alarm_type}")

		# Redis 可用性检查
		if self.redis is None:
			logger.warning("[RedisRelay] Redis不可用，直接推送事件")
			self._direct_broadcast(event_type, payload)
			return

		try:
			# 原子性设置key
			set_result = self.redis.set(key, "1", nx=True, ex=ttl_seconds)

			if set_result:
				# 成功设置key，这是新事件
				logger.info(f"[RedisRelay] 新事件，开始推送: type={event_type}, alarm_type={alarm_type}")
				self._direct_broadcast(event_type, payload)
			else:
				# SET NX失败，key已存在，进行检查
				key_exists = self.redis.exists(key)
				ttl_remaining = self.redis.ttl(key)

				logger.warning(f"[RedisRelay] 检测到重复事件: type={event_type}, alarm_type={alarm_type}, key_exists={key_exists}, ttl={ttl_remaining}")

				# 处理竞态条件
				if key_exists and ttl_remaining > 0:
					if ttl_remaining < 10:  # TTL小于10秒，可能是清理延迟
						logger.warning(f"[RedisRelay] key即将过期，强制推送: type={event_type}, alarm_type={alarm_type}")
						self._direct_broadcast(event_type, payload)
					else:
						# 确实是重复事件
						logger.info(f"[RedisRelay] 重复事件被跳过: type={event_type}, alarm_type={alarm_type}, ttl={ttl_remaining}")
				else:
					# 竞态条件，强制推送
					logger.warning(f"[RedisRelay] Redis竞态条件，强制推送: type={event_type}, alarm_type={alarm_type}")
					self._direct_broadcast(event_type, payload)
		except Exception as e:
			logger.error(f"[RedisRelay] Redis操作失败，降级处理: {e}", exc_info=True)
			# Redis失败时仍然要进行WebSocket推送
			self._direct_broadcast(event_type, payload)

	@staticmethod
	def _direct_broadcast(event_type: str, payload: Dict[str, Any]) -> None:
		"""直接广播事件"""
		try:
			# 获取告警信息用于日志
			alarm_type = payload.get('alarm_type', 'UNKNOWN')
			meter_code = payload.get('meter_code', 'UNKNOWN')
			alarm_id = payload.get('alarm_id', 'UNKNOWN')

			# 增强日志：记录所有事件的广播尝试
			logger.info(f"[RedisRelay] 准备广播事件: type={event_type}, alarm_type={alarm_type}, meter_code={meter_code}, alarm_id={alarm_id}")

			# 根据事件类型选择推送方式
			from app.routers.dashboard_router import broadcast_new_alert
			import asyncio
			import threading

			# 异步广播函数
			async def async_broadcast():
				try:
					logger.info(f"[RedisRelay] 开始执行WebSocket广播: event_type={event_type}")
					if event_type == EventType.ALARM_RAISED.value:
						logger.info(f"[RedisRelay] 广播新告警: {meter_code} - {alarm_type}")
						await broadcast_new_alert(payload, "new")
					elif event_type == EventType.ALARM_RECOVERED.value:
						logger.info(f"[RedisRelay] 广播告警恢复: {meter_code} - {alarm_type}")
						await broadcast_new_alert(payload, "recovered")
					elif event_type == EventType.WORK_ORDER_CLOSED.value:
						# 工单关闭事件不向前端推送
						logger.info(f"[RedisRelay] 工单关闭事件，不推送: {meter_code}")
					else:
						logger.warning(f"[RedisRelay] 未知事件类型: {event_type}")
						await broadcast_new_alert(payload, event_type.lower())
					logger.info(f"[RedisRelay] WebSocket广播完成: event_type={event_type}")
				except Exception as e:
					logger.error(f"[RedisRelay] WebSocket广播失败: {e}", exc_info=True)

			# 处理事件循环
			try:
				loop = asyncio.get_running_loop()
				logger.info(f"[RedisRelay] 检测到运行中的事件循环，创建任务")
				# 关键修复：使用asyncio.ensure_future并立即等待，确保任务被执行
				# 注意：在同步线程中不能直接await，所以使用run_coroutine_threadsafe
				future = asyncio.run_coroutine_threadsafe(async_broadcast(), loop)
				# 等待执行完成（最多等待10秒）
				try:
					future.result(timeout=10)
					logger.info(f"[RedisRelay] 任务执行完成: event_type={event_type}")
				except Exception as e:
					logger.error(f"[RedisRelay] 任务执行失败: {e}", exc_info=True)
			except RuntimeError:
				# 没有运行的事件循环，创建新的
				logger.info(f"[RedisRelay] 未检测到运行中的事件循环，创建新线程")
				def run_in_thread():
					try:
						new_loop = asyncio.new_event_loop()
						asyncio.set_event_loop(new_loop)
						new_loop.run_until_complete(async_broadcast())
					except Exception as e:
						logger.error(f"[RedisRelay] 广播失败: {e}", exc_info=True)
					finally:
						try:
							new_loop.close()
						except Exception:
							pass
						asyncio.set_event_loop(None)

				thread = threading.Thread(target=run_in_thread, daemon=True)
				thread.start()
				# 等待线程执行完成（最多10秒）
				thread.join(timeout=10)
				if thread.is_alive():
					logger.warning(f"[RedisRelay] 广播线程超时")
				else:
					logger.info(f"[RedisRelay] 新线程广播完成")
			except Exception as e:
				logger.error(f"[RedisRelay] 创建广播任务失败: {e}", exc_info=True)

		except Exception as e:
			logger.error(f"[RedisRelay] 事件广播处理失败: {e}", exc_info=True)


class KafkaRelay(MessageRelay):
	"""预留：下月实现"""

	def relay(self, event_type: str, payload: Dict[str, Any]) -> None:
		# TODO: 初始化 kafka producer 并投递
		logger.warning("[KafkaRelay] 未实现，事件丢弃 %s", event_type)
